package com.tbit.uqbike.webmanager.controller.interceptor.stat;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author MyWin E-mail:335918956@qq.com
 * @version 1.0
 * @createTime 2019-11-14 20:23
 */
@Slf4j
public class DefStatServices implements StatServices, Runnable {

    public static final int MQ_MSG_ID_PLATFORM_RECORD = 50004;
    public static final String pointDataStatKey = "point_data";


    private String host;
    private Integer port;
    private String exchange;
    private String username;
    private String password;
    private RabbitTemplate rabbitTemplate;
    private ConnectionFactory connectionFactory;
    private List<PlatformRecord> platformRecordList = new LinkedList<>();
    private ReentrantLock lock = new ReentrantLock();

    public DefStatServices() {
        // 启动推送线程
        new Thread(this).start();
    }

    private static String builtPushMsg(List<PlatformRecord> temp) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("msgId", MQ_MSG_ID_PLATFORM_RECORD);
        jsonObject.put("data", new PlatformRecordData(temp));
        return jsonObject.toJSONString();
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public String getExchange() {
        return exchange;
    }

    public void setExchange(String exchange) {
        this.exchange = exchange;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    @Override
    public void recordStatData(String comName, String statName, String statValue, Date dt, String exData) {
        PlatformRecord pr = new PlatformRecord(comName, statName, dt, statValue, exData);
        try {
            lock.lock();
            platformRecordList.add(pr);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void run() {
        while (true) {
            // 每隔1s取一次数据，批量推送过去
            List<PlatformRecord> temp;
            try {
                lock.lock();
                temp = platformRecordList;
                platformRecordList = new LinkedList<>();
            } finally {
                lock.unlock();
            }
            if (!temp.isEmpty()) {
                try {
                    pushMsg(pointDataStatKey, builtPushMsg(temp));
                } catch (Exception e) {
                    log.error("推送日志记录失败:", e);
                }
            }
            // 不管有没有休眠1s
            try {
                Thread.sleep(1 * 1000);
            } catch (InterruptedException e) {

            }
        }
    }

    private ConnectionFactory getConnectionFactory() {
        if (null == connectionFactory) {
            CachingConnectionFactory ccf = new CachingConnectionFactory();
            ccf.setHost(host);
            ccf.setPort(port);
            ccf.setUsername(username);
            ccf.setPassword(password);

            connectionFactory = ccf;
        }
        return connectionFactory;
    }

    private RabbitTemplate getRabbitTemplate() {
        if (rabbitTemplate == null) {

            ConnectionFactory cf = getConnectionFactory();
            if (null != cf) {
                rabbitTemplate = new RabbitTemplate();
                rabbitTemplate.setConnectionFactory(cf);
                rabbitTemplate.setExchange(exchange);
            }
        }
        return rabbitTemplate;
    }

    private void pushMsg(String key, String msg) throws UnsupportedEncodingException {
        RabbitTemplate rabbitTemplate = getRabbitTemplate();
        if (null != rabbitTemplate) {
            rabbitTemplate.send(key, new Message(msg.getBytes("UTF-8"), new org.springframework.amqp.core.MessageProperties()));
        }
    }
}
